package com.zhengb.flink.study.datastream.connector;

import com.zhengbo.flink.demo.base.constant.FlinkConstant;
import com.zhengbo.flink.demo.base.kafka.KafkaConfig;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;

import java.util.Properties;

/**
 * Created by zhengbo on 2020/2/1.
 */
public class FlinkKafkaProducerTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        String topic = "neu_car";

        Properties properties = KafkaConfig.kafkaConsumerProperties(FlinkConstant.KAFKA_BROKER_LIST, "flink-test");

        FlinkKafkaProducer011<String> kafkaProducer011 =
                new FlinkKafkaProducer011<>(topic,
                        new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), properties);

        env.socketTextStream("10.0.10.36", 9999).addSink(kafkaProducer011);


        env.execute("FlinkKafkaProducerTest");
    }
}
